# middleware/redis_adapter.py
"""Redis适配器：提供统一的Redis客户端接口，支持单节点、哨兵、集群模式"""

from typing import Optional, Dict, Union, Any
from loguru import logger
from utils.logger import global_logger

from utils.logger import Logger
from utils.exceptions import RedisError
from utils.constants import RedisMode
from config import RedisClusterConfig, load_redis_config
from middleware.redis_client import RedisClient  # 原有的单节点客户端
from middleware.redis_cluster_client import RedisClusterClient  # 新的集群客户端


class RedisAdapter:
    """Redis适配器：统一的Redis客户端接口"""
    
    _instance = None
    _client = None
    _config = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(RedisAdapter, cls).__new__(cls)
            cls._instance._init_client()
        return cls._instance
    
    def _init_client(self):
        """初始化Redis客户端"""
        try:
            # 加载配置
            self._config = load_redis_config()
            # logger.info(f"初始化Redis适配器，模式: {self._config.mode}")
            
            # 根据模式选择客户端
            if self._config.mode == RedisMode.SINGLE:
                # 单节点模式，使用原有的RedisClient
                self._client = RedisClient()
                global_logger.success("单节点Redis客户端初始化成功")
            elif self._config.mode in [RedisMode.SENTINEL, RedisMode.CLUSTER]:
                # 哨兵或集群模式，使用新的RedisClusterClient
                self._client = RedisClusterClient(self._config)
                global_logger.success(f"{self._config.mode} Redis客户端初始化成功")
            else:
                raise ValueError(f"不支持的Redis模式: {self._config.mode}")
                
        except Exception as e:
            global_logger.error(f"Redis适配器初始化失败: {str(e)}")
            raise RuntimeError(f"Redis适配器初始化失败: {str(e)}") from e
    
    def get_client(self):
        """获取实际的Redis客户端实例"""
        return self._client
    
    def get_config(self) -> RedisClusterConfig:
        """获取Redis配置"""
        return self._config
    
    # ==================== 代理方法：转发到实际客户端 ====================
    
    def get(self, key: str) -> Optional[Union[str, bytes]]:
        """获取缓存值"""
        return self._client.get(key)
    
    def set(self, key: str, value: Union[str, bytes], ex: Optional[int] = None) -> bool:
        """设置缓存"""
        return self._client.set(key, value, ex)
    
    def delete(self, key: str) -> int:
        """删除缓存"""
        return self._client.delete(key)
    
    def exists(self, key: str) -> int:
        """检查key是否存在"""
        return self._client.exists(key)
    
    def expire(self, key: str, seconds: int) -> bool:
        """设置过期时间"""
        return self._client.expire(key, seconds)
    
    def ttl(self, key: str) -> int:
        """获取剩余过期时间"""
        return self._client.ttl(key)
    
    def setex(self, key: str, seconds: int, value: Union[str, bytes]) -> bool:
        """设置带过期时间的缓存值（set with expire）"""
        return self.set(key, value, ex=seconds)
    
    def pipeline(self):
        """创建Redis事务管道"""
        if hasattr(self._client, 'pipeline'):
            return self._client.pipeline()
        else:
            global_logger.warning(f"客户端不支持pipeline操作，返回None")
            return None
            
    def zremrangebyscore(self, key, min_score, max_score):
        """删除有序集合中分数在指定范围内的成员"""
        try:
            if hasattr(self._client, 'zremrangebyscore'):
                return self._client.zremrangebyscore(key, min_score, max_score)
            else:
                global_logger.warning(f"客户端不支持zremrangebyscore操作")
                return 0
        except Exception as e:
            global_logger.error(f"执行zremrangebyscore失败: {str(e)}")
            return 0
            
    def zcard(self, key):
        """获取有序集合的成员数量"""
        try:
            if hasattr(self._client, 'zcard'):
                return self._client.zcard(key)
            else:
                global_logger.warning(f"客户端不支持zcard操作")
                return 0
        except Exception as e:
            global_logger.error(f"执行zcard失败: {str(e)}")
            return 0
            
    def zadd(self, key, mapping):
        """向有序集合添加成员"""
        try:
            if hasattr(self._client, 'zadd'):
                return self._client.zadd(key, mapping)
            else:
                global_logger.warning(f"客户端不支持zadd操作")
                return 0
        except Exception as e:
            global_logger.error(f"执行zadd失败: {str(e)}")
            return 0
    
    # ==================== Hash操作 ====================
    
    def hget(self, key: str, field: str) -> Optional[Union[str, bytes]]:
        """获取Hash字段值"""
        return self._client.hget(key, field)
    
    def hset(self, key: str, field: str, value: Union[str, bytes]) -> bool:
        """设置Hash字段值"""
        return self._client.hset(key, field, value)
    
    def hgetall(self, key: str) -> Dict[str, str]:
        """获取所有Hash字段和值"""
        return self._client.hgetall(key)
    
    def hdel(self, key: str, field: str) -> int:
        """删除Hash字段"""
        if hasattr(self._client, 'hdel'):
            return self._client.hdel(key, field)
        else:
            # 对于单节点客户端，可能需要实现hdel
            global_logger.warning(f"客户端不支持hdel操作，使用替代方案")
            return self._delete_hash_field(key, field)
    
    def _delete_hash_field(self, key: str, field: str) -> int:
        """删除Hash字段的替代实现"""
        try:
            # 获取当前所有字段
            data = self.hgetall(key)
            if field in data:
                del data[field]
                # 重新设置整个Hash
                self._client.delete(key)
                if data:
                    self.hsetall(key, data)
                return 1
            return 0
        except Exception as e:
            global_logger.error(f"删除Hash字段失败: {str(e)}")
            return 0
    
    def hsetall(self, key: str, mapping: Dict[str, Union[str, bytes]]) -> bool:
        """批量设置Hash字段值"""
        return self._client.hsetall(key, mapping)
    
    # ==================== List操作 ====================
    
    def lpush(self, key: str, *values) -> int:
        """左推入列表（同步接口，适配异步客户端）"""
        try:
            if hasattr(self._client, 'lpush'):
                # 区分同步和异步客户端
                import inspect
                if inspect.iscoroutinefunction(self._client.lpush):
                    # 对于异步客户端，使用事件循环执行
                    import asyncio
                    try:
                        # 尝试获取已运行的事件循环
                        loop = asyncio.get_event_loop()
                        if loop.is_running():
                            # 在运行中的事件循环中，使用线程执行
                            import threading
                            result = [0]
                            error = [None]
                            
                            def sync_call():
                                try:
                                    inner_loop = asyncio.new_event_loop()
                                    asyncio.set_event_loop(inner_loop)
                                    result[0] = inner_loop.run_until_complete(self._client.lpush(key, *values))
                                except Exception as e:
                                    error[0] = e
                                finally:
                                    inner_loop.close()
                            
                            thread = threading.Thread(target=sync_call)
                            thread.start()
                            thread.join(timeout=5)  # 5秒超时
                            
                            if error[0]:
                                raise error[0]
                            return result[0]
                        else:
                            # 事件循环存在但未运行，直接运行
                            return loop.run_until_complete(self._client.lpush(key, *values))
                    except RuntimeError:
                        # 没有事件循环，创建新的
                        return asyncio.run(self._client.lpush(key, *values))
                else:
                    # 同步客户端直接调用
                    return self._client.lpush(key, *values)
            else:
                global_logger.warning(f"客户端不支持lpush操作")
                return 0
        except Exception as e:
            global_logger.error(f"执行lpush失败: {str(e)}")
            return 0
    
    def rpush(self, key: str, *values) -> int:
        """右推入列表（同步接口，适配异步客户端）"""
        try:
            if hasattr(self._client, 'rpush'):
                # 区分同步和异步客户端
                import inspect
                if inspect.iscoroutinefunction(self._client.rpush):
                    # 对于异步客户端，使用事件循环执行
                    import asyncio
                    try:
                        # 尝试获取已运行的事件循环
                        loop = asyncio.get_event_loop()
                        if loop.is_running():
                            # 在运行中的事件循环中，使用线程执行
                            import threading
                            result = [0]
                            error = [None]
                            
                            def sync_call():
                                try:
                                    inner_loop = asyncio.new_event_loop()
                                    asyncio.set_event_loop(inner_loop)
                                    result[0] = inner_loop.run_until_complete(self._client.rpush(key, *values))
                                except Exception as e:
                                    error[0] = e
                                finally:
                                    inner_loop.close()
                            
                            thread = threading.Thread(target=sync_call)
                            thread.start()
                            thread.join(timeout=5)  # 5秒超时
                            
                            if error[0]:
                                raise error[0]
                            return result[0]
                        else:
                            # 事件循环存在但未运行，直接运行
                            return loop.run_until_complete(self._client.rpush(key, *values))
                    except RuntimeError:
                        # 没有事件循环，创建新的
                        return asyncio.run(self._client.rpush(key, *values))
                else:
                    # 同步客户端直接调用
                    return self._client.rpush(key, *values)
            else:
                global_logger.warning(f"客户端不支持rpush操作")
                return 0
        except Exception as e:
            global_logger.error(f"执行rpush失败: {str(e)}")
            return 0
    
    def lpop(self, key: str) -> Optional[str]:
        """左弹出列表（同步接口，适配异步客户端）"""
        try:
            if hasattr(self._client, 'lpop'):
                # 区分同步和异步客户端
                import inspect
                if inspect.iscoroutinefunction(self._client.lpop):
                    # 对于异步客户端，使用事件循环执行
                    import asyncio
                    try:
                        # 尝试获取已运行的事件循环
                        loop = asyncio.get_event_loop()
                        if loop.is_running():
                            # 在运行中的事件循环中，使用线程执行
                            import threading
                            result = [None]
                            error = [None]
                            
                            def sync_call():
                                try:
                                    inner_loop = asyncio.new_event_loop()
                                    asyncio.set_event_loop(inner_loop)
                                    result[0] = inner_loop.run_until_complete(self._client.lpop(key))
                                except Exception as e:
                                    error[0] = e
                                finally:
                                    inner_loop.close()
                            
                            thread = threading.Thread(target=sync_call)
                            thread.start()
                            thread.join(timeout=5)  # 5秒超时
                            
                            if error[0]:
                                raise error[0]
                            return result[0]
                        else:
                            # 事件循环存在但未运行，直接运行
                            return loop.run_until_complete(self._client.lpop(key))
                    except RuntimeError:
                        # 没有事件循环，创建新的
                        return asyncio.run(self._client.lpop(key))
                else:
                    # 同步客户端直接调用
                    return self._client.lpop(key)
            else:
                global_logger.warning(f"客户端不支持lpop操作，使用替代实现")
                # 替代实现：获取第一个元素并删除
                try:
                    # 获取第一个元素
                    value = self._client.lindex(key, 0)
                    if value is not None:
                        # 删除第一个元素
                        self._client.ltrim(key, 1, -1)
                    return value
                except Exception as e:
                    global_logger.error(f"执行lpop替代方案失败: {str(e)}")
                    return None
        except Exception as e:
            global_logger.error(f"执行lpop失败: {str(e)}")
            return None
    
    def rpop(self, key: str) -> Optional[str]:
        """右弹出列表（同步接口，适配异步客户端）"""
        try:
            if hasattr(self._client, 'rpop'):
                # 区分同步和异步客户端
                import inspect
                if inspect.iscoroutinefunction(self._client.rpop):
                    # 对于异步客户端，使用事件循环执行
                    import asyncio
                    try:
                        # 尝试获取已运行的事件循环
                        loop = asyncio.get_event_loop()
                        if loop.is_running():
                            # 在运行中的事件循环中，使用线程执行
                            import threading
                            result = [None]
                            error = [None]
                            
                            def sync_call():
                                try:
                                    inner_loop = asyncio.new_event_loop()
                                    asyncio.set_event_loop(inner_loop)
                                    result[0] = inner_loop.run_until_complete(self._client.rpop(key))
                                except Exception as e:
                                    error[0] = e
                                finally:
                                    inner_loop.close()
                            
                            thread = threading.Thread(target=sync_call)
                            thread.start()
                            thread.join(timeout=5)  # 5秒超时
                            
                            if error[0]:
                                raise error[0]
                            return result[0]
                        else:
                            # 事件循环存在但未运行，直接运行
                            return loop.run_until_complete(self._client.rpop(key))
                    except RuntimeError:
                        # 没有事件循环，创建新的
                        return asyncio.run(self._client.rpop(key))
                else:
                    # 同步客户端直接调用
                    return self._client.rpop(key)
            else:
                global_logger.warning(f"客户端不支持rpop操作，使用替代实现")
                # 替代实现：使用Lua脚本或其他方式模拟rpop
                return self._fallback_rpop(key)
        except Exception as e:
            global_logger.error(f"执行rpop失败: {str(e)}")
            return None
            
    def _fallback_rpop(self, key: str) -> Optional[str]:
        """rpop的替代实现"""
        # 注意：这是一个简化的实现，不保证原子性
        try:
            # 获取列表长度
            length = self.llen(key)
            if length == 0:
                return None
                
            # 检查客户端是否支持lindex和ltrim操作
            if not hasattr(self._client, 'lindex'):
                global_logger.warning(f"客户端不支持lindex操作，无法实现rpop替代")
                return None
                
            # 获取最后一个元素
            value = self._client.lindex(key, -1)
            if value is not None and hasattr(self._client, 'ltrim'):
                # 删除最后一个元素
                self._client.ltrim(key, 0, length - 2)
            elif value is not None:
                global_logger.warning(f"客户端不支持ltrim操作，已获取元素但无法删除")
            return value
        except Exception as e:
            global_logger.error(f"执行rpop替代方案失败: {str(e)}")
            return None
    
    def llen(self, key: str) -> int:
        """获取列表长度（同步接口，适配异步客户端）"""
        try:
            if hasattr(self._client, 'llen'):
                # 区分同步和异步客户端
                import inspect
                if inspect.iscoroutinefunction(self._client.llen):
                    # 使用统一的异步工具函数处理
                    from utils.async_utils import run_async_in_sync
                    try:
                        return run_async_in_sync(self._client.llen, key)
                    except Exception as e:
                        global_logger.error(f"异步执行llen失败: {str(e)}")
                        # 降级到替代方案
                        raise
                else:
                    # 同步客户端直接调用
                    return self._client.llen(key)
            else:
                global_logger.warning(f"客户端不支持llen操作，使用替代实现")
                # 替代实现：先检查客户端是否支持lrange
                try:
                    if hasattr(self._client, 'lrange'):
                        # 使用lrange获取所有元素并计算长度
                        # 注意：这对大列表可能性能较差
                        elements = self._client.lrange(key, 0, -1)
                        return len(elements)
                    else:
                        # 如果客户端不支持lrange，直接返回0
                        global_logger.warning(f"客户端不支持lrange操作，无法获取列表长度，返回0")
                        return 0
                except Exception as e:
                    global_logger.error(f"执行llen替代方案失败: {str(e)}")
                    return 0
        except Exception as e:
            global_logger.error(f"执行llen失败: {str(e)}")
            return 0
    
    # ==================== Set操作 ====================
    
    def sadd(self, key: str, *values) -> int:
        """添加集合成员"""
        if hasattr(self._client, 'sadd'):
            return self._client.sadd(key, *values)
        else:
            global_logger.warning(f"客户端不支持sadd操作")
            return 0
    
    def smembers(self, key: str) -> set:
        """获取集合所有成员"""
        if hasattr(self._client, 'smembers'):
            return self._client.smembers(key)
        else:
            global_logger.warning(f"客户端不支持smembers操作")
            return set()
    
    def scard(self, key: str) -> int:
        """获取集合基数"""
        if hasattr(self._client, 'scard'):
            return self._client.scard(key)
        else:
            global_logger.warning(f"客户端不支持scard操作")
            return 0
    
    def srem(self, key: str, *values) -> int:
        """移除集合成员"""
        if hasattr(self._client, 'srem'):
            return self._client.srem(key, *values)
        else:
            global_logger.warning(f"客户端不支持srem操作")
            return 0
    
    # ==================== 分布式锁 ====================
    
    def acquire_lock(self, lock_name: str, timeout: int = 30) -> bool:
        """获取分布式锁"""
        return self._client.acquire_lock(lock_name, timeout)
    
    def release_lock(self, lock_name: str) -> bool:
        """释放分布式锁"""
        return self._client.release_lock(lock_name)
    
    def is_locked(self, lock_name: str) -> bool:
        """检查锁是否被持有"""
        if hasattr(self._client, 'is_locked'):
            return self._client.is_locked(lock_name)
        else:
            # 对于单节点客户端，检查锁是否存在
            global_logger.warning(f"客户端不支持is_locked操作")
            return self.exists(f"lock:{lock_name}") > 0
    
    # ==================== 发布订阅 ====================
    
    def publish(self, channel: str, message: str) -> int:
        """发布消息"""
        return self._client.publish(channel, message)
    
    def subscribe(self, channel: str, callback, stop_event):
        """订阅频道（支持异步客户端）"""
        try:
            # 检查客户端类型（异步还是同步）
            import inspect
            if hasattr(self._client, 'subscribe') and inspect.iscoroutinefunction(self._client.subscribe):
                # 异步客户端，在单独的线程中运行事件循环
                import threading
                import asyncio
                
                def subscription_thread():
                    """订阅线程函数"""
                    loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(loop)
                    
                    async def subscribe_async():
                        """异步订阅函数"""
                        try:
                            pubsub = await self._client.subscribe(channel)
                            self.logger.info(f"成功订阅频道: {channel}")
                            
                            # 持续监听消息
                            while not stop_event.is_set():
                                message = await pubsub.get_message(ignore_subscribe_messages=True)
                                if message:
                                    try:
                                        # 解析消息并调用回调
                                        data = message.get('data')
                                        if isinstance(data, str):
                                            import json
                                            data = json.loads(data)
                                        callback(data)
                                    except Exception as e:
                                        self.logger.error(f"处理订阅消息失败: {str(e)}")
                                # 短暂休眠避免CPU占用过高
                                await asyncio.sleep(0.1)
                        except Exception as e:
                            self.logger.error(f"异步订阅失败: {str(e)}")
                        finally:
                            if 'pubsub' in locals():
                                try:
                                    await pubsub.unsubscribe(channel)
                                    await pubsub.close()
                                except:
                                    pass
                    
                    # 运行异步订阅函数
                    loop.run_until_complete(subscribe_async())
                    loop.close()
                
                # 启动订阅线程
                thread = threading.Thread(target=subscription_thread, daemon=True)
                thread.start()
                self.logger.info(f"启动异步订阅线程，频道: {channel}")
                return thread
            else:
                # 同步客户端
                # 注意：RedisClient.subscribe方法需要三个参数
                # 这里我们创建一个简单的回调函数和停止事件
                import threading
                from utils.logger import Logger
                
                # 创建logger（如果没有的话）
                if not hasattr(self, 'logger'):
                    self.logger = Logger.get_logger(agent_name="redis_adapter")
                    
                # 创建停止事件
                stop_event = threading.Event()
                
                def simple_callback(data):
                    """简单回调函数，将消息传递给原始回调"""
                    if data:
                        callback(data)
                
                # 启动一个单独的线程来调用subscribe方法
                subscribe_thread = threading.Thread(
                    target=self._client.subscribe,
                    args=(channel, simple_callback, stop_event),
                    daemon=True
                )
                subscribe_thread.start()
                self.logger.info(f"成功启动订阅线程，频道: {channel}")
                
                # 保存订阅线程和停止事件，以便后续可以取消订阅
                self._subscriptions = getattr(self, '_subscriptions', {})
                self._subscriptions[channel] = (subscribe_thread, stop_event)
                return subscribe_thread
        except Exception as e:
            # 确保有logger可用
            if not hasattr(self, 'logger'):
                from utils.logger import Logger
                self.logger = Logger.get_logger(agent_name="redis_adapter")
                
            self.logger.error(f"订阅频道失败: {str(e)}")
            raise
    
    def unsubscribe(self, *channels):
        """取消订阅"""
        return self._client.unsubscribe(*channels)
    
    # ==================== 健康检查 ====================
    
    def ping(self) -> bool:
        """健康检查"""
        try:
            if hasattr(self._client, 'ping'):
                return self._client.ping()
            else:
                # 通过执行简单命令检查
                self._client.get("health_check")
                return True
        except Exception as e:
            logger.error(f"Redis健康检查失败: {str(e)}")
            return False
    
    # ==================== 性能监控 ====================
    
    def get_info(self) -> Dict[str, Any]:
        """获取Redis信息"""
        info = {
            "mode": self._config.mode,
            "connected": self.ping(),
            "config": {
                "host": self._config.host,
                "port": self._config.port,
                "db": self._config.db if hasattr(self._config, 'db') else None,
                "pool_size": self._config.pool_size if hasattr(self._config, 'pool_size') else None
            }
        }
        
        # 尝试获取更详细的信息
        try:
            if hasattr(self._client, 'get_info'):
                client_info = self._client.get_info()
                info.update(client_info)
        except Exception as e:
            logger.warning(f"获取详细Redis信息失败: {str(e)}")
        
        return info


# ==================== 全局实例 ====================

def get_redis_adapter() -> RedisAdapter:
    """获取Redis适配器全局实例"""
    return RedisAdapter()

def get_redis_client():
    """获取Redis客户端（兼容原有接口）"""
    adapter = get_redis_adapter()
    return adapter.get_client()